package com.at.bigdata.spark.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/19 10:02 PM
 */
object Spark06_Req1_PageflowAnalysis {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("operator")
    val sc = new SparkContext(sparkConf)

    // 1、读取原始数据
    val actionRDD = sc.textFile("datas/user_visit_action.txt")

    val actionDataRDD = actionRDD.map(
      action => {
        val datas = action.split("_")
        new UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong
        )
      }
    )
    actionDataRDD.cache()

    // TODO 对指定的连续页面跳转进行统计
    //
    val ids = List(1L, 2L, 3L, 4L, 6L, 7L)
    val okflowIds = ids.zip(ids.tail)

    // TODO 计算分母
    val page2CountMap = actionDataRDD.filter(
      iter => {
        ids.init.contains(iter.page_id)
      }
    ).map(
      action => {
        (action.page_id, 1L)
      }
    ).reduceByKey(_ + _).collect().toMap
    // TODO 计算分子
    // 根据session进行分组
    val sessionRDD = actionDataRDD.groupBy(_.session_id)
    // 分组后，根据访问的时间进行排序
    val mvRDD = sessionRDD.mapValues(
      iter => {
        val sortList = iter.toList.sortBy(_.action_time)
        val flowIds = sortList.map(_.page_id)
        // [1,2,3,4]
        // [1,2],[2,3],[3,4]
        // [1-2, 2-3, 3-4]
        // Sliding: 滑窗
        // [1,2,3,4]
        // [2,3,4]
        // 拉链
        val pageflowIds = flowIds.zip(flowIds.tail)
        // 将不合法的页面跳转数据进行不合法过滤
        pageflowIds.filter(
          t => {
            okflowIds.contains(t)
          }
        ).map(
          t => {
            (t, 1L)
          }
        )
      }
    )
    // ((1,2), 1)
    val flatRDD = mvRDD.map(_._2).flatMap(list => list)
    // ((1,2), 1) => ((1,2), sum)
    val dataRDD = flatRDD.reduceByKey(_ + _)

    // TODO 计算转化率
    // 分子除以分母
    dataRDD.foreach {
      case ((pageid1, pageid2), sum) => {
        val lon = page2CountMap.getOrElse(pageid1, 0L)
        println(s"页面${pageid1}跳转到页面${pageid2}单跳转换率为：" + sum.toString)
      }
    }

    sc.stop()
  }

  case class UserVisitAction(
                              date: String, // 用户点击行为的日期
                              user_id: Long, // 用户的Id
                              session_id: String, // session的id
                              page_id: Long, // 某个页面的id
                              action_time: String, // 动作的时间点
                              search_keyword: String, // 用户搜索的关键字
                              click_category_id: Long, // 某一个商品品类id
                              click_product_id: Long, // 某一个商品的id
                              order_category_ids: String, // 一次订单中所有品类的id集合
                              order_product_ids: String, // 一次订单中所有商品的id集合
                              pay_category_ids: String, // 一次支付中所有的品类的id集合
                              pay_product_ids: String, // 一次支付中所有的商品的id集合
                              city_id: Long // 城市id
                            )
}
